package top.wintp.offlinedataanalysis.anlyser.mr.nu;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import top.wintp.offlinedataanalysis.anlyser.dim.StatsUserDimension;
import top.wintp.offlinedataanalysis.anlyser.value.map.TimeOutputValue;
import top.wintp.offlinedataanalysis.anlyser.value.reduce.MapWritableValue;
import top.wintp.offlinedataanalysis.common.KpiType;

/**
 * description: reducer 去重 整合数据
 * <p>
 *
 * @author: upuptop
 * <p>
 * qq: 337081267
 * <p>
 * CSDN:   http://blog.csdn.net/pyfysf
 * <p>
 * cnblogs:   http://www.cnblogs.com/upuptop
 * <p>
 * blog:   http://wintp.top
 * <p>
 * email:  pyfysf@163.com
 * <p>
 * time: 2019/08/2019/8/23
 * <p>
 */
public class NewInstallUserReducer extends Reducer<StatsUserDimension, TimeOutputValue, StatsUserDimension, MapWritableValue> {
    private static final Logger logger = LoggerFactory.getLogger(NewInstallUserReducer.class);


    /**
     * 保存用户的id ——uuid
     * 为了去重的效果使用set集合
     */
    private Set<String> uuidSets = new HashSet<>();
    /**
     * 创建out - value
     */
    private MapWritableValue mapWritableValue = new MapWritableValue();

    /**
     * out-value-key
     */
    private IntWritable mapWritableValueKey = new IntWritable(-1);
    /**
     * out-value-value
     */
    private IntWritable mapWritableValueValue = new IntWritable();

    @Override
    protected void reduce(StatsUserDimension key, Iterable<TimeOutputValue> values, Context context) throws IOException, InterruptedException {
        uuidSets.clear();

        //   迭代数据
        for (TimeOutputValue value : values) {
            String uuid = value.getId();
            //    对用户id进行去重
            uuidSets.add(uuid);
        }
        //    保存用户的个数
        MapWritable mapValue = new MapWritable();
        mapWritableValueValue.set(uuidSets.size());
        mapValue.put(mapWritableValueKey, mapWritableValueValue);

        //        设置输出的value
        mapWritableValue.setValue(mapValue);
        mapWritableValue.setKpi(KpiType.valueOfName(key.getStatsCommon().getKpi().getKpiName()));
        //    写出数据
        context.write(key, mapWritableValue);
    }
}
